package com.jz.pay.consumer.topicListener;

import com.jz.pay.bo.RocketMQBO;
import com.jz.pay.consumer.TopicType;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageClientIDSetter;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;

@Component
public class UserListerner  implements MessageListenerConcurrently {
    @Resource
    RocketMQBO rocketMQBO;
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext queue){
        MessageQueue messageQueue = queue.getMessageQueue();
        System.out.println("----------------------------------------------------------");
        System.out.println("本次接收："+messageQueue.getBrokerName()+" "+messageQueue.getTopic()+" "+msgs.size());
        for (int i = 0; i < msgs.size(); i++) {
            MessageExt msg = msgs.get(i);
            String body = new String(msg.getBody());
            System.out.println(messageQueue.getBrokerName()+","+messageQueue.getTopic()+"   "+ body);
            if(!run(msg,body)){
                sendDelay(msg,body);
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
    private boolean run(MessageExt messageExt,String body){
        // 执行逻辑
        boolean isDelay = body.contains("delay");
        if (isDelay){
            System.out.println("进入补偿队列："+body+" "+messageExt.getMsgId());
        }else if(runned(messageExt)){
            System.out.println("重复消费："+body+" "+messageExt.getMsgId());
        }else {
            System.out.println("正常执行："+body+" "+messageExt.getMsgId());
        }
        return !isDelay;
    }
    private void sendDelay(MessageExt messageExt,String body){
        HashMap<String,String> map = new HashMap<>();
        map.put(RocketMQHeaders.KEYS,messageExt.getKeys());
        map.put(RocketMQHeaders.TAGS,messageExt.getMsgId());
        rocketMQBO.send(TopicType.LOG.getRetry(), body,map);
    }

    static HashSet<String> runnedMsgs = new HashSet<>();
    private boolean runned(MessageExt messageExt){
        return !runnedMsgs.add(MessageClientIDSetter.getUniqID(messageExt));
    }
}
